package demo1;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.util.Arrays;

public class WordCountJob {

    public static void main(String[] args) throws Exception {
        System.out.println(Arrays.toString(args));
        // 1.创建配置类对象
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://hadop10:9000");
        // 2.创建一个Job对象
        Job job = Job.getInstance(conf);
        job.setJarByClass(WordCountJob.class);
        // 3.设置输入和输出的类
        job.setInputFormatClass(TextInputFormat.class);
        job.setOutputFormatClass(TextOutputFormat.class);
        // 4.设置读取的文件路径和输出的路径
//        TextInputFormat.addInputPath(job,new Path("/wc.txt"));
//        TextOutputFormat.setOutputPath(job,new Path("/out2"));  //该目录不能存在
        TextInputFormat.addInputPath(job,new Path(args[0]));
        TextOutputFormat.setOutputPath(job,new Path(args[1]));  //该目录不能存在
        // 5.设置自定义的Mapper类和Reducer类
        job.setMapperClass(WordCountMapper.class);
        job.setReducerClass(WordCountReducer.class);
        // 6.设置Mapper输出的key-value类型和Reducer输出的key-value的类型
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IntWritable.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(IntWritable.class);
        // 7.提交作业，等待完成，参数true代表打印运行的日志信息
        boolean result = job.waitForCompletion(true);
        System.out.println(result);  // 返回true代表运行成功
    }

    //Mapper<输入key的类型，输入value的类型，输出key的类型，输出value的类型>
    //LongWritable、Text、IntWritable都是hadoop的内置数据类型
    //LongWritable  long
    //Text          String
    // text对象.toString()  将Text类型转换成String
    // Text text = new Text(字符串)  将字符串转换成Text
    // IntWritable  int
    // int i = intWritable对象.get()                             将IntWritable转换成int类型
    // IntWritable intWritable = new IntWritable(Int数值)        将int数值转换成IntWritable类型
    static class WordCountMapper extends Mapper<LongWritable, Text,Text, IntWritable> {
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //参数为MapTask输入的key 和 value
            //key(偏移量)   value(行数据)
            //  0          zhangsan lisi wangwu

            // 1、将行数据按照指定分隔符进行分割
            String[] arr = value.toString().split(" ");
            // 2、遍历数组，获取每个姓名
            for (String name : arr) {
                Text text = new Text(name);
                IntWritable intWritable = new IntWritable(1);
                context.write(text,intWritable);
            }
        }
    }

    static class WordCountReducer extends Reducer<Text,IntWritable,Text,IntWritable>{
        @Override
        protected void reduce(Text key, Iterable<IntWritable> values, Context context) throws IOException, InterruptedException {
            // 思考输入的key value
            //         zs  [1,1,1,1,1]
            int sum = 0;
            for (IntWritable v : values) {
                int value = v.get();  // 将IntWritable对象中存储的int值获取出来
                sum += value;
            }
            context.write(key,new IntWritable(sum));
            // 输出的key value
            //      zs   5
        }
    }
}
